/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.namesrv.kvconfig;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map.Entry;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.remoting.protocol.body.KVTable;

/**
 * 当前类的作用
 *
 * 1. 管理kv配置
 * 2. 管理kv配置的读写锁
 */
public class KVConfigManager {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);

    /**
     * 命名服务器控制器实例
     */
    private final NamesrvController namesrvController;

    /**
     * 读写锁
     */
    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    /**
     * key是命名空间，
     * value是HashMap，以kv形式记录配置信息
     */
    private final HashMap<String/* Namespace */, HashMap<String/* Key */, String/* Value */>> configTable =
        new HashMap<>();

    public KVConfigManager(NamesrvController namesrvController) {
        this.namesrvController = namesrvController;
    }

    /**
     * 加载配置信息：<br>
     * 解析配置文件为配置键值对，添加到configTable中。
     * configTable的key是命名空间名称，value是HashMap（以键值对形式记录的配置信息）
     */
    public void load() {
        String content = null;
        try {
            // 将配置文件的配置信息解析为字符串
            content = MixAll.file2String(this.namesrvController.getNamesrvConfig().getKvConfigPath());
            // 配置文件的路径是这个
            // private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
        } catch (IOException e) {
            log.warn("Load KV config table exception", e);
        }
        if (content != null) {
            // 将配置json字符串反序列化为KVConfigSerializeWrapper对象
            KVConfigSerializeWrapper kvConfigSerializeWrapper =
                KVConfigSerializeWrapper.fromJson(content, KVConfigSerializeWrapper.class);
            if (null != kvConfigSerializeWrapper) {
                // 如果有配置条目，则将配置信息添加到configTable中。
                this.configTable.putAll(kvConfigSerializeWrapper.getConfigTable());
                log.info("load KV config table OK");
            }
        }
    }

    /**
     * 将指定的配置键值对添加到指定命名空间中。<br>
     * 为了保证并发的数据安全，使用读写锁保护。
     * @param namespace 命名空间
     * @param key   配置的key
     * @param value 配置的value
     */
    public void putKVConfig(final String namespace, final String key, final String value) {
        try {
            // 获取写锁
            this.lock.writeLock().lockInterruptibly();
            try {
                // 从configTable中获取指定命名空间的配置信息的HashMap集合
                HashMap<String, String> kvTable = this.configTable.get(namespace);
                // 如果没有找到，则创建新的HashMap
                if (null == kvTable) {
                    kvTable = new HashMap<>();
                    // 将命名空间作为key，新的键值对Map集合作为value放到configTable中
                    this.configTable.put(namespace, kvTable);
                    log.info("putKVConfig create new Namespace {}", namespace);
                }

                // 将指定的配置键值对信息添加到kvTable中
                final String prev = kvTable.put(key, value);
                // 如果原来就有，则表示覆盖了原来的配置
                if (null != prev) {
                    log.info("putKVConfig update config item, Namespace: {} Key: {} Value: {}",
                        namespace, key, value);
                } else {
                    log.info("putKVConfig create new config item, Namespace: {} Key: {} Value: {}",
                        namespace, key, value);
                }
            } finally {
                // 释放写锁
                this.lock.writeLock().unlock();
            }
        } catch (InterruptedException e) {
            log.error("putKVConfig InterruptedException", e);
        }

        // 持久化
        this.persist();
    }

    /**
     * 配置信息的持久化
     */
    public void persist() {
        try {
            // 获取读锁，防止并发的写
            this.lock.readLock().lockInterruptibly();
            try {
                // 创建KVConfigSerializeWrapper对象
                KVConfigSerializeWrapper kvConfigSerializeWrapper = new KVConfigSerializeWrapper();
                // 设置配置表
                kvConfigSerializeWrapper.setConfigTable(this.configTable);

                // 将配置信息序列化为json字符串
                String content = kvConfigSerializeWrapper.toJson();

                if (null != content) {
                    // 如果有配置信息，则将配置信息写到配置文件中
                    MixAll.string2File(content, this.namesrvController.getNamesrvConfig().getKvConfigPath());
                }
            } catch (IOException e) {
                log.error("persist kvconfig Exception, "
                    + this.namesrvController.getNamesrvConfig().getKvConfigPath(), e);
            } finally {
                // 释放读锁
                this.lock.readLock().unlock();
            }
        } catch (InterruptedException e) {
            log.error("persist InterruptedException", e);
        }

    }

    /**
     * 删除配置键值对。
     * @param namespace 配置所属的命名空间（用于隔离集群）
     * @param key 要删除的配置的key。
     */
    public void deleteKVConfig(final String namespace, final String key) {
        try {
            // 获取写锁
            this.lock.writeLock().lockInterruptibly();
            try {
                // 根据命名空间获取配置键值对表
                HashMap<String, String> kvTable = this.configTable.get(namespace);
                if (null != kvTable) {
                    // 如果配置表存在，则从配置表中删除指定的键值对配置
                    String value = kvTable.remove(key);
                    log.info("deleteKVConfig delete a config item, Namespace: {} Key: {} Value: {}",
                        namespace, key, value);
                }
            } finally {
                // 释放写锁
                this.lock.writeLock().unlock();
            }
        } catch (InterruptedException e) {
            log.error("deleteKVConfig InterruptedException", e);
        }

        // 执行配置信息的持久化，写到配置文件中。
        this.persist();
    }

    /**
     * 获取指定命名空间的配置表
     * @param namespace 指定的命名空间名称
     * @return 返回配置表的字节数组
     */
    public byte[] getKVListByNamespace(final String namespace) {
        try {
            // 获取读锁
            this.lock.readLock().lockInterruptibly();
            try {
                // 获取指定命名空间的配置表
                HashMap<String, String> kvTable = this.configTable.get(namespace);
                if (null != kvTable) {
                    // 如果配置表存在，则创建KVTable对象
                    KVTable table = new KVTable();
                    // 设置配置表
                    table.setTable(kvTable);
                    // 将配置表编码为字节数组并返回
                    return table.encode();
                }
            } finally {
                // 释放读锁
                this.lock.readLock().unlock();
            }
        } catch (InterruptedException e) {
            log.error("getKVListByNamespace InterruptedException", e);
        }
        // 否则返回null值
        return null;
    }

    /**
     * 获取指定命名空间，指定key的配置值
     * @param namespace 命名空间名称
     * @param key       配置条目的key
     * @return 返回配置条目的value
     */
    public String getKVConfig(final String namespace, final String key) {
        try {
            // 获取读锁
            this.lock.readLock().lockInterruptibly();
            try {
                // 获取指定命名空间的配置表
                HashMap<String, String> kvTable = this.configTable.get(namespace);
                if (null != kvTable) {
                    // 如果配置表存在，则返回指定key的value。
                    return kvTable.get(key);
                }
            } finally {
                // 释放读锁
                this.lock.readLock().unlock();
            }
        } catch (InterruptedException e) {
            log.error("getKVConfig InterruptedException", e);
        }

        // 如果找不到，则返回null值
        return null;
    }

    /**
     * 周期性打印所有配置信息
     */
    public void printAllPeriodically() {
        try {
            // 获取读锁
            this.lock.readLock().lockInterruptibly();
            try {
                log.info("--------------------------------------------------------");

                {
                    log.info("configTable SIZE: {}", this.configTable.size());
                    for (Entry<String, HashMap<String, String>> next : this.configTable.entrySet()) {
                        for (Entry<String, String> nextSub : next.getValue().entrySet()) {
                            log.info("configTable NS: {} Key: {} Value: {}", next.getKey(), nextSub.getKey(),
                                    nextSub.getValue());
                        }
                    }
                }
            } finally {
                // 打印完成后释放读锁
                this.lock.readLock().unlock();
            }
        } catch (InterruptedException e) {
            log.error("printAllPeriodically InterruptedException", e);
        }
    }
}
